/**
 * 
 */
package com.cnhaimei.vod.net.dq;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Hashtable;
import java.util.concurrent.ConcurrentLinkedQueue;

import android.os.Handler;
import android.os.Message;
import android.util.Log;

import com.cnhaimei.com.VODLog;
import com.cnhaimei.net.NetSocket;
import com.cnhaimei.net.TCPSocket;
import com.cnhaimei.net.NetSocket.SocketDataListener;
import com.cnhaimei.vod.logic.EventsManager;
import com.cnhaimei.vod.net.DQHeart;
import com.cnhaimei.vod.net.Data;
import com.cnhaimei.vod.net.Request;
import com.cnhaimei.vod.util.Util;

/**
 * voD歌曲、歌星数据库查询 <br>
 * 
 * @author jin001.wu@gmail.com <br>
 * 
 */
public class DQManager extends TCPSocket implements SocketDataListener {
	public static final long IDENTIFIER_FAIL = -1;
	private static long IDENTIFIER = System.currentTimeMillis(); // 记数器
	public static final String TAG = "DQManager";
	public static final String Tab_SONG = "song_valid";
	public static final String Tab_SINGER = "star_valid";
	private static final int MSG_TIME_KEEP = 1;
	private static final int HEART_TIME = 20 * 1000;
	public static final byte SNAP_FAIL = 0;

	public static final int PORT = 2006;
	static String DQIp = "10.0.0.9";//"10.0.0.9";
	
	// /**客户登录*/
	// public static final byte LOGIN = 0x01;// (暂不支持)
	/** 断开连接 */
	public static final byte DISCONNECT = 0x02;
	/** 建立查询 */
	public static final byte SEARCH = 0x03;
	/** 重新查询 */
	public static final byte RESEARCH = 0x04;
	/** 关闭查询 */
	public static final byte CLOSE = 0x05;
	/** 读取记录 */
	public static final byte READ = 0x06;
	// /**更新记录*/
	// public static final byte UPDATE = 0x07;// (暂不支持)
	// /**排序记录*/
	// public static final byte SORT = 0x08;// (暂不支持)
	/** 辅助控制 */
	public static final char SUPCONTROL = 0x09;
	public static final int HEAD_SIZE = 4;

	public static final byte Zero = 0;
	public final ByteArrayOutputStream ByteOStream = new ByteArrayOutputStream(
			1024);
	public static final byte[] INPUT_BYTE = new byte[48 * 1024];

	public static final String ENCODING = "UTF-8";

	private final ConcurrentLinkedQueue<Long> QUEUE = new ConcurrentLinkedQueue<Long>();

	Hashtable<Long, Parser> parserArray = new Hashtable<Long, Parser>();
	// SparseArray<Parser> parserArray;
	HeartKeeper mKeeper;
	RequestQueue mRequestQueue = new RequestQueue();

	public DQManager() {
		super(DQIp, PORT, new byte[HEAD_SIZE]);
		this.setSocketDataListener(this);
		mKeeper = new HeartKeeper();
		// Queue<E>
	}
	
	//自定义ip
	public DQManager(String strIp) {
		super(strIp, PORT, new byte[HEAD_SIZE]);
		this.setSocketDataListener(this);
		mKeeper = new HeartKeeper();
	}
	

	@Override
	protected void onBeforeConnect() {
		String ip = null;

		if (getIp() != null) {
			return;
		}
			
		while (ip == null) {
			ip = DQHeart.getVODServerIP();
			if (ip == null) {
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			} else {
				setIp(ip);
			}
		}
	}

	@Override
	protected void onConnected(boolean isSuc) {
		if (Util.VLOG) {
			VODLog.e(TAG,"dqmanager onConnected");
		}

		Log.e(TAG,"dqmanager onConnected");
		QUEUE.clear();
		mRequestQueue.cleanRequest();
		EventsManager.G().sendEvent(EventsManager.Event.CONNECTING, id, true);
		autoKeepHeart();
	}

	@Override
	protected void onDisConnected() {
		Log.e(TAG, "disconnect");
		if (Util.VLOG) {
			VODLog.e(TAG, "dqmanager ondisconnect");
		}
		mRequestQueue.cleanRequest();
		mKeeper.removeMessages(MSG_TIME_KEEP);
		EventsManager.G().sendEvent(EventsManager.Event.DISCONNECT, id, false);
	}
	
	public void close() {
		if (mKeeper != null) {
			mKeeper.removeMessages(MSG_TIME_KEEP);
		}
		super.close();
	}

	public void data(NetSocket socket, byte[] head, int headLenth) {
		// Log.e(TAG, "data parse");
		if (Util.VLOG) {
			VODLog.e(TAG, "dqmanager data parse" + head[0]);
		}
		DQHead dqHead = new DQHead(head[0], head[1], head[2], head[3]);
		parseData(socket, dqHead.com, dqHead);
	}

	private void parseData(NetSocket socket, byte com, DQHead head) {
		// Log.e(TAG, "parse");
		if (head.snap == SNAP_FAIL) {
			// TODO
			// Log.e(TAG, "fail");
			// Log.e(TAG, " response snap: " + head.snap);
			// return;
		}

		switch (com) {

		case DISCONNECT: {

		}
			break;
		case SEARCH: {
			Log.e(TAG, "response query data");
			responseQuery(head);
			mRequestQueue.pollRequest();
			if (Util.VLOG) {
				VODLog.e(TAG, "poll request with search.");
			}
		}
			break;
		case RESEARCH: {

		}
			break;
		case CLOSE: {
			Log.e(TAG, "resonpse close snap:" + head.snap);
			responseClose(head);
			mRequestQueue.pollRequest();
			if (Util.VLOG) {
				VODLog.e(TAG, "poll request with close.");
			}
		}
			break;
		case READ: {
			Log.e(TAG, "response read data");
			responseData(head);
			mRequestQueue.pollRequest();
			if (Util.VLOG) {
				VODLog.e(TAG, "poll request with read.");
			}
		}
			break;
		default:
			break;
		}

	}

	public long addRequest(DQRequest req) {
		if (req == null) {
			return IDENTIFIER_FAIL;
		}
		long id = genAndPutId();
		req.setId(id);
		mRequestQueue.addRequest(req);
		if (Util.VLOG) {
			VODLog.e(TAG, "add request: " + req.description() + " id" + id);
		}
		return id;
	}

	/**
	 * 发送查询请求包
	 * 
	 * @param tableName
	 * @param query
	 */
	public void requestQuery(String tableName, String query) {
		byte xcmmd = SEARCH;
		byte xargv = 0;
		boolean isSuc = false;
		try {
			if (Util.VLOG) {
				VODLog.e(TAG, "dqmanager requestQuery tableName:" + tableName
						+ " query:" + query);
			}
			
			byte svar[] = new byte[32];
			byte[] nameBytes = tableName.getBytes(ENCODING);
			System.arraycopy(nameBytes, 0, svar, 0, nameBytes.length);

			byte[] queryBtyes = query.getBytes(ENCODING);
			
			byte endByte = 0;

			byte hSize = (byte) ((queryBtyes.length + 1 + 32) >> 8);
			byte lSize = (byte) ((queryBtyes.length + 1 + 32) >> 0);

			final ByteArrayOutputStream byteOut = ByteOStream;
			byteOut.reset();
			byteOut.write(xcmmd);
			byteOut.write(xargv);
			byteOut.write(hSize);
			byteOut.write(lSize);
			byteOut.write(svar);
			byteOut.write(queryBtyes);
			byteOut.write(endByte);

			byteOut.writeTo(mOutput);
			isSuc = true;
		} catch (Exception e) {
			e.printStackTrace();
			mRequestQueue.pollRequest(); // 失败弹出
			if (Util.VLOG) {
				VODLog.e(TAG, "poll request with error");
			}
		}
		
		EventsManager.G().sendEvent(EventsManager.Event.QUERY_QUERY, null,
				isSuc);
	}

	/**
	 * 读取记录
	 * 
	 * @param snap
	 * @param readFrom
	 * @param readCount
	 */
	void requestData(Parser parser, int readFrom, int readCount) {

		DQParser sp = (DQParser) parser;
		byte xcmmd = READ;
		byte xargv = sp.getSnap();// 接收查询应答包接收到的xhead.xargv
		byte hSize = 0;
		byte lSize = 8;

		final ByteArrayOutputStream byteOut = new ByteArrayOutputStream();// ByteOStream;
		byteOut.reset();
		byteOut.write(xcmmd);
		byteOut.write(xargv);
		byteOut.write(hSize);
		byteOut.write(lSize);
		try {
			if (Util.VLOG) {
				VODLog.e(TAG, "dqmanager requestQuery readFrom:" + readFrom
						+ " readCount:" + readCount);
			}
			
			writeInt(readFrom, byteOut);
			writeInt(readCount, byteOut);
			// byteOut.write(readFrom); // 从查询结果的第1条记录开始读取
			// byteOut.write(readCount);// 读取352条记录

			byteOut.writeTo(mOutput);
			// rtn = genAndPutId();
			// parserArray.put(rtn, sp);
			byteOut.flush();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			mRequestQueue.pollRequest(); // 失败弹出
			if (Util.VLOG) {
				VODLog.e(TAG, "poll request with error");
			}
		}
		// return rtn;
	}

	/**
	 * 关闭快照
	 * 
	 * @param snap
	 */
	void requestClose(byte snap) {

		byte xcmmd = 5;
		byte xargv = snap;// 接收读取数据的回应包xhead.xargv
		byte hSize = 0;
		byte lSize = 0;

		final ByteArrayOutputStream byteOut = ByteOStream;
		byteOut.reset();
		byteOut.write(xcmmd);
		byteOut.write(xargv);
		byteOut.write(hSize);
		byteOut.write(lSize);

		try {
			if (Util.VLOG) {
				VODLog.e(TAG, "dqmanager close snap:" + snap);
			}
			byteOut.writeTo(mOutput);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			mRequestQueue.pollRequest(); // 失败弹出
		}
	}

	private void requestKeep() {
		byte xcmmd = 9;
		byte xargv = 0;
		byte hSize = 0;
		byte lSize = 0;

		final ByteArrayOutputStream byteOut = ByteOStream;
		byteOut.reset();
		byteOut.write(xcmmd);
		byteOut.write(xargv);
		byteOut.write(hSize);
		byteOut.write(lSize);

		try {
			if (Util.VLOG) {
				VODLog.e(TAG, "dqmanager requestKeep");
			}
			byteOut.writeTo(mOutput);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	protected void responseQuery(DQHead head) {
		DQQueryResult response = new DQQueryResult();
		boolean isSuc = false;

		try {
			
			DQRequest request = mRequestQueue.getRunningRequest();
			if (request instanceof DQQuery) {
				response.setRequest(request);
			}
			Log.e(TAG, "snap:" + head.snap);

			Arrays.fill(INPUT_BYTE, Zero);
			mInput.read(INPUT_BYTE, 0, head.size);
			ByteArrayInputStream in = new ByteArrayInputStream(INPUT_BYTE);
			response.setSnap(head.snap);
			isSuc = head.snap != 0 ? true : false;
			if (isSuc) {
				isSuc = response.parse(in, head.size);
			}

			
			if (request instanceof DQQuery) {
				request.setState(Request.STATE_RESPONSED);
				((DQQuery)request).setSnap(head.snap);
				request.onResponsed();
			}

		} catch (Exception e) {
			// TODO
			e.printStackTrace();
		}

		long id = pollId();
		response.setId(id);
		EventsManager.G().sendEvent(EventsManager.Event.QUERY_RESPONSE, response,
				isSuc);
	}

	protected void responseData(DQHead head) {
		// Log.e(TAG, "responseData");
		boolean isSuc = false;
		Data data = null;

		long id = pollId();

		try {
//			Log.e(TAG, "snap:" + head.snap);
			Arrays.fill(INPUT_BYTE, Zero);
			mInput.read(INPUT_BYTE, 0, head.size);
			DQRequest request = mRequestQueue.getRunningRequest();
			if (request instanceof DQRead) {
				DQRead read = (DQRead) request;
				Parser parser = read.getParser();
				if (parser != null) {
					data = parser.parse(INPUT_BYTE, head);
				}
				request.setState(Request.STATE_RESPONSED);
			}

			isSuc = true;

		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		parserArray.remove(id);
		if (data != null) {
			data.setId(id);
		}
		EventsManager.G().sendEvent(EventsManager.Event.QUERY_RECORDS, data,
				isSuc);
	}

	private void responseClose(DQHead head) {
		byte snap = head.snap;
		long id = pollId(); // is need
		boolean isSuc = false;
		DQRequest request = mRequestQueue.getRunningRequest();
		if (request instanceof DQClose) {
			DQClose close = (DQClose) request;
			close.setState(Request.STATE_RESPONSED);
		}

		isSuc = head.snap != 0 ? true : false;
		EventsManager.G().sendEvent(EventsManager.Event.QUERY_CLOSE, snap,
				isSuc);
	}

	public void stop() {
		if (mKeeper != null) {
			mKeeper.removeMessages(MSG_TIME_KEEP);
		}
		if (mInput != null) {
			try {
				mInput.close();
			} catch (Exception e) {
			}
			mInput = null;
		}
		if (mOutput != null) {
			try {
				mOutput.close();
			} catch (Exception e) {
			}
			mOutput = null;
		}
		if (mSocket != null) {
			super.close();
			mSocket = null;
		}
	}

	private void autoKeepHeart() {
		Log.e(TAG, "keepHeart");
		requestKeep();
		mKeeper.removeMessages(MSG_TIME_KEEP);
		mKeeper.sendEmptyMessageDelayed(MSG_TIME_KEEP, HEART_TIME);
	}

	public void start() {
		if (!isRun()) {
			Thread t = new Thread(this);
			t.start();
		}
	}

	private long genAndPutId() {
		IDENTIFIER++;
		if (IDENTIFIER <= 0L) {
			IDENTIFIER = 1;
		}
		QUEUE.add(IDENTIFIER);
		return IDENTIFIER;
	}

	protected long pollId() {
		Long l = QUEUE.poll();
		return l == null ? 0L : l;
	}

	public static short readShort(InputStream iStream) {
		short rtn = 0;
		try {

			rtn = (short) ((iStream.read() & 0xff) << 8 + (iStream.read() & 0xff));
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return rtn;
	}

	public static String readString(byte[] bs) {
		int index = bs.length;
		for (int i = 0; i < bs.length; i++) {
			if (bs[i] == 0) {
				index = i;
				break;
			}
		}

		String rtn = "";

		try {
			rtn = new String(bs, 0, index, ENCODING);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}

		return rtn;
	}

	public static void writeInt(int v, OutputStream os) throws IOException {
		os.write((byte) v);
		os.write((byte) (v >>> 8));
		os.write((byte) (v >>> 16));
		os.write((byte) (v >>> 24));

		// os.write((byte)(v >>> 24));
		// os.write((byte)(v >>> 16));
		// os.write((byte)(v >>> 8));
		// os.write((byte) v);

	}

	private class HeartKeeper extends Handler {

		@Override
		public void handleMessage(Message msg) {
			switch (msg.what) {
			case MSG_TIME_KEEP:
				autoKeepHeart();
				break;

			default:
				break;
			}
			super.handleMessage(msg);
		}
	}

	private class RequestQueue {
		private final Object lock = new Object();
		private ArrayDeque<DQRequest> waiting = new ArrayDeque<DQRequest>(10);
		private DQRequest running;

		public void addRequest(DQRequest req) {
			synchronized (lock) {
				waiting.add(req);
				Log.e(TAG, "waiting queue:" + waiting.size());

				if (running == null) {
					startRequest();
				}
			}
		}

		public synchronized void cleanRequest() {
			synchronized (lock) {
				running = null;
				waiting.clear();
			}
		}

		public synchronized void pollRequest() {
			synchronized (lock) {
				running = null;
				startRequest();
			}
		}

		public synchronized DQRequest getRunningRequest() {
			synchronized (lock) {
				return running;
			}
		}

		private void startRequest() {
			DQRequest req = waiting.poll();
			if (req != null) {
				req.setState(Request.STATE_RUNNING);
				req.doRequest(DQManager.this);
				running = req;
			}
		}
	}
}
